package com.gitee.ywj1352.reactive;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class ZKSampleSubscription implements Subscription {

    private DecoratingSubscriber subscriber;

    CuratorFramework client;

    public ZKSampleSubscription(Subscriber subscriber) {
        this.subscriber = new DecoratingSubscriber(subscriber);

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.newClient("192.168.128.129:2181",
                5000, 5000, retryPolicy);
        client.start();

    }

    @Override
    public void request(long n) {
        if (n < 1 || n > 5) {
            throw new IllegalArgumentException("The number of elements to requests must be more than zero!");
        }
        System.out.println("SampleSubscription#request" + n);
        subscriber.setMaxRequest(n - 1);

        try {
            client.create().creatingParentContainersIfNeeded() // 递归创建所需父节点
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL) // 创建类型为持久节点
                    .forPath("/event/", "init".getBytes()); // 目录及内容




        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    @Override
    public void cancel() {
        subscriber.onComplete();
        System.out.println("SampleSubscription#cancel:" + subscriber.getClass());
    }
}
